package com.shujia.utils

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.{Logger, LoggerFactory}



/**
  * spark 工具类
  * spark代码继承这个类，实现run方法
  *
  */
abstract class SparkTool {

  var sc: SparkContext = _
  var log: Logger = _
  var conf: SparkConf = _

  def main(args: Array[String]): Unit = {

    //初始化打日志的对象
    log = LoggerFactory.getLogger(this.getClass)
    conf = new SparkConf()
    conf.setAppName(this.getClass.getName)

    //指定配置
    init()

    sc = new SparkContext(conf)


    //调用子类的方法
    run(args)


  }

  /**
    * 子类实现此方法，实现具体的代码逻辑
    *
    */
  def run(args: Array[String]): Unit

  /**
    * 指定spark配置
    * conf.set(key,value)
    */

  def init(): Unit

  /**
    * 将rdd的数据保存到hdfs,  rdd的类型必须是String类型
    *
    */
  def save(rdd: RDD[String], path: String): Unit = {
    /**
      * 保存数据到文件
      */
    //删除输出目录,  使用hadoop的方式删除
    val config = new Configuration()
    val fileSystem = FileSystem.get(config)
    if (fileSystem.exists(new Path(path))) {
      fileSystem.delete(new Path(path), true)
    }
    //保存位置统合表
    rdd.saveAsTextFile(path)
  }

}
